package com.allenjsl.springactuator.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.util.CollectionUtils;

import java.util.List;

@Slf4j
public class OrderMessageListener implements MessageListenerOrderly {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {


        if (CollectionUtils.isEmpty(msgs)){
            return ConsumeOrderlyStatus.SUCCESS;
        }
        //设置自动提交
        context.setAutoCommit(true);
        msgs.stream().parallel()
                .forEach(msg -> {
                    try {
                        String messageBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        log.info("Handle Order Message: messageId: " + msg.getMsgId() + ",topic: " + msg.getTopic() + ",tags: "
                                + msg.getTags() + ",keys: " + msg.getKeys() + ",messageBody: " + messageBody);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
